/*
 * KeyValuePageFixedKeyVariantValueImpl.h
 *
 *  Created on: Jul 24, 2011
 *      Author: yong
 */

#ifndef KEYVALUEPAGEFIXEDKEYVARIANTVALUEIMPL__H_
#define KEYVALUEPAGEFIXEDKEYVARIANTVALUEIMPL__H_
#include "../Page.h"
#include "../MemPageMaster.h"
#include "../Utility.h"
#include "../KeyValue.h"
#include "../Typedef.h"
#include "../Mapper.h"
#include "../Callable.h"
#include "../traits/is_fixed_size.h"
#include <boost/type_traits.hpp>
#include <boost/utility.hpp>
#include <iostream>
#include <cstring>
#include <cstdio>
#include <vector>
#include <string>
#include <algorithm>
using namespace std;
namespace distrim
{

/**
 *
 */
template<typename K, typename V>
class KeyValuePage<K, V, true, false>
{
public:
	KeyValuePage(size_t pageID, /*MemPageMaster<K, V> *ptrMemPageMaster,*/
			void *ptrPageBeg, const size_t pairsCount,
			const size_t maxPairCount = MAX_DEFAULT_PAIR_COUNT,
			const size_t maxPageLenBytes =
					MAX_DEFAULT_KEY_VALUE_PAGE_LENTH_BYTES) :
		Page<K, V> (pageID, ptrPageBeg/*, ptrMemPageMaster*/), m_maxPairCount(
				maxPairCount), m_maxPageLenBytes(maxPageLenBytes),
				m_pairsCount(pairsCount), m_orderState(OrderState::None)
	{
		initializePointers();
	}
	// Default constructor, for new[] operator to create array of KeyValuePage records.
	KeyValuePage(const size_t maxPairCount = MAX_DEFAULT_PAIR_COUNT,
			const size_t maxPageLenBytes =
					MAX_DEFAULT_KEY_VALUE_PAGE_LENTH_BYTES) :
		Page<K, V> (0, NULL), m_maxPairCount(maxPairCount),
				m_maxPageLenBytes(maxPageLenBytes), m_pairsCount(0),
				m_orderState(OrderState::None), m_ptrSecIndexes(NULL),
				m_ptrSecKeyValues(NULL)
	{
	}
	virtual ~KeyValuePage()
	{
	}

private:

	/**
	 * Maximum count of key-pairs.
	 */
	size_t m_maxPairCount;
	/**
	 * Maximum page length in bytes.
	 */
	size_t m_maxPageLenBytes;
	/**
	 * Number of key-value pairs within this page.
	 */
	size_t m_pairsCount;
	/**
	 * Pairs sorted or not.
	 */
	OrderState::OrderStateEnum m_orderState;
	/**
	 * Beginning of key-value indexes in memory, containing key-value pairs' beginning
	 * offset.
	 */
	IndexWithLengthLessThan<K*, size_t> *m_ptrSecIndexes;
	/**
	 * The real content of key-value pairs: key 1, value 1,
	 * key 2, value2...key n, value n.
	 */
	K *m_ptrSecKeyValues;

	/**
	 * The functor which filles this page.
	 */
	Callable m_producerClass;
	/**
	 * The functor which consumes this page.
	 */
	Callable m_consumerClass;
	/**
	 * Precondition: Number of max pairs within this page must be known.
	 */
	inline void initializePointers()
	{
		// pair counts, key bytes, value element bytes, value element count, page state.
		m_ptrSecIndexes
				= (IndexWithLengthLessThan<K*, size_t>*) Page<K, V>::m_ptrPageBeg;
		m_ptrSecKeyValues
				= (K*) ((char*) m_ptrSecIndexes
						+ sizeof(IndexWithLengthLessThan<K*, size_t> )
								* m_maxPairCount);
	}

public:

	/**
	 * Set the producer class.
	 */
	inline void SetProducerClass(const Callable &producer)
	{
		m_producerClass = producer;
	}
	/**
	 * Set the cosumer class.
	 */
	inline void SetConsumerClass(const Callable &consumer)
	{
		m_consumerClass = consumer;
	}
	/**
	 *	Set maximum pairs this page can accommadate.
	 */
	inline void SetMaxPairCount(const size_t maxPairCount)
	{
		m_maxPairCount = maxPairCount;
	}
	/**
	 * Get maximum pairs this page can accommadate.
	 */
	inline const size_t GetMaxPairCount() const
	{
		return m_maxPairCount;
	}
	/**
	 * Set maximum page length in bytes.
	 */
	inline void SetMaxPageLength(const size_t maxPageLenBytes)
	{
		m_maxPageLenBytes = maxPageLenBytes;
	}
	/**
	 * Get maximum page length in bytes.
	 */
	inline const size_t GetMaxPageLength() const
	{
		return m_maxPageLenBytes;
	}
	/**
	 * Get number of pairs within this page.
	 */
	inline const size_t GetPairsCount() const
	{
		return m_pairsCount;
	}
	/**
	 *
	 */
	inline OrderState::OrderStateEnum GetOrderSate() const
	{
		return m_orderState;
	}
	/**
	 *
	 */
	inline void SetOrderState(OrderState::OrderStateEnum &orderState)
	{
		m_orderState = orderState;
	}
	/**
	 * @precodition: number of max pair count must be known.
	 */
	inline void SetPageBegin(void *begPtr)
	{
		Page<K, V>::pSetPageBegin(begPtr);
		initializePointers();
	}

	/**
	 *	Define how key-value pairs within this page be written to disk in binary mode
	 *	as a continuous chunk.
	 */
	virtual void WriteToDisk(const string &fileName)
	{
		if (fileName == string(""))
		{
			throw std::invalid_argument("no file to write");
		}
		if (!Page<K, V>::m_storageState != StorageState::Filled)
		{
			throw std::logic_error("page has not been filled");
		}
		if (!Page<K, V>::m_isWrittenToDisk)
		{
			FILE *pf;
			pf = fopen(fileName.c_str(), "wb");
			if (pf == NULL)
			{
				throw std::runtime_error("error open file to write");
			}
			// page bytes, max pair counts, pair counts,
			//	key bytes, value element bytes, value element count, page order state.
			fwrite(&m_maxPairCount, 1, sizeof(size_t), pf);
			fwrite(&m_maxPageLenBytes, 1, sizeof(size_t), pf);
			fwrite(&m_pairsCount, 1, sizeof(size_t), pf);
			fwrite(&m_orderState, 1, sizeof(OrderState::OrderStateEnum), pf);
			// Write valid index offset.
			ptrdiff_t offset = 0;
			size_t length = 0;
			for (size_t i = 0; i < m_pairsCount; ++i)
			{
				// Write offset of key-value pair and value length in bytes
				//	to file.
				offset = (char*) m_ptrSecIndexes[i].PtrIndex() - (char*) Page<
						K, V>::m_ptrPageBeg;
				length = m_ptrSecIndexes[i].Length();
				fwrite(&offset, 1, sizeof(K*), pf);
				fwrite(&length, 1, sizeof(size_t), pf);
			}
			// Write content of key-values.
			for (size_t i = 0; i < m_pairsCount; ++i)
			{
				fwrite(m_ptrSecIndexes[i].PtrIndex(), sizeof(char), sizeof(K)
						+ m_ptrSecIndexes[i].Length(), pf);
			}
			fclose(pf);
			Page<K, V>::m_isWrittenToDisk = true;
		}
	}

	/**
	 * Define how key-value pairs be restored from a continuous chunk in disk in binary mode.
	 */
	virtual void ReadFromDisk(const string &fileName)
	{
		if (fileName == string(""))
		{
			throw std::logic_error("no file name to read");
		}
		if (Page<K, V>::m_storageState == StorageState::Empty)
		{
			throw std::logic_error("page has not been allocated");
		}
		if (Page<K, V>::m_isWrittenToDisk && Page<K, V>::m_storageState
				== StorageState::Allocated)
		{
			FILE *pf;
			pf = fopen(fileName.c_str(), "rb");
			if (pf == NULL)
			{
				throw std::runtime_error("error open file to read");
			}
			// page bytes, pair counts, key bytes, value element bytes, value element count, page state.
			fread(&m_maxPairCount, sizeof(size_t), 1, pf);
			fread(&m_maxPageLenBytes, sizeof(size_t), 1, pf);
			fread(&m_pairsCount, sizeof(size_t), 1, pf);
			fread(&m_orderState, sizeof(OrderState::OrderStateEnum), 1, pf);
			// Read address of key-value pair contents' offsets and value length in bytes in.
			fread(Page<K, V>::m_ptrPageBeg, sizeof(IndexWithLengthLessThan<K*,
					size_t> ), m_pairsCount, pf);
			m_ptrSecIndexes
					= (IndexWithLengthLessThan<K*, size_t> *) Page<K, V>::m_ptrPageBeg;
			m_ptrSecKeyValues = (K*) ((char*) m_ptrSecIndexes
					+ sizeof(IndexWithLengthLessThan<K*, size_t> )
							* m_maxPairCount);
			size_t kvBytes = 0;
			for (size_t i = 0; i < m_pairsCount; ++i)
			{
				// Sum up all key-value pairs' content length in bytes.
				kvBytes += (sizeof(K) + m_ptrSecIndexes[i].Length());
			}
			// Read content of key-value pairs in.
			fread(m_ptrSecKeyValues, sizeof(char), kvBytes, pf);
			fclose(pf);
			// Add offsets with obsolute address to get real address.
			for (size_t i = 0; i < m_pairsCount; ++i)
			{
				m_ptrSecIndexes[i].PtrIndex(
						(K*) ((char*) Page<K, V>::m_ptrPageBeg
								+ (ptrdiff_t) m_ptrSecIndexes[i].PtrIndex()));
			}
			Page<K, V>::m_storageState = StorageState::Filled;
		}
	}

	/**
	 * Combine the key-value pairs within this page, none
	 * new page created.
	 */
	virtual void Combine()
	{
		if (Page<K, V>::m_storageState == StorageState::Empty)
		{
			throw std::logic_error("page has not been allocated");
		}
		if (m_orderState != OrderState::Combined)
		{
			if (m_orderState == OrderState::None)
			{
				Sort();
			}
			std::vector<IndexWithLengthLessThan<K*, size_t> > diffPtrs;
			KeyValue<K, V> kv;
			KeyValue<K, V> tmp;
			// Combine values with same key.
			for (size_t i = 0; i < m_pairsCount;)
			{
				// Record the address of first key-pair with different key.
				diffPtrs.push_back(m_ptrSecIndexes[i]);
				kv.SetKeyPtr(m_ptrSecIndexes[i].PtrIndex());
				kv.SetValueBegPtr((V*) ((char*) m_ptrSecIndexes[i].PtrIndex()
						+ sizeof(K)));
				kv.SetValueElemCount(m_ptrSecIndexes[i].Length());
				while (++i < m_pairsCount)
				{
					tmp.SetKeyPtr(m_ptrSecIndexes[i].PtrIndex());
					tmp.SetValueBegPtr(
							(V*) ((char*) m_ptrSecIndexes[i].PtrIndex()
									+ sizeof(K)));
					tmp.SetValueElemCount(m_ptrSecIndexes[i].Length());
					if (kv.GetKey() == tmp.GetKey())
					{
						kv + tmp;
					}
					else
					{
						break;
					}
				}
			}
			// Update pairs count.
			m_pairsCount = diffPtrs.size();
			// Sort the valid pointers in ascending address in case of overlapping
			//	during compression.
			std::sort(diffPtrs.begin(), diffPtrs.end());
			// Clean old indexes.
			memset(m_ptrSecIndexes, 0, sizeof(IndexWithLengthLessThan<K*,
					size_t> ) * m_maxPairCount);
			// Collect and compress valid indexes,
			//	move key-paris forward to compress.
			size_t offset = 0;
			for (size_t i = 0; i < m_pairsCount; ++i)
			{
				diffPtrs[i].PtrIndex((K*) ((char*) diffPtrs[i].PtrIndex()
						+ offset));
				m_ptrSecIndexes[i].PtrIndex(
						(K*) ((char*) diffPtrs[i].PtrIndex() + offset));
				m_ptrSecIndexes[i].Length(diffPtrs[i].Length());
				memcpy(m_ptrSecIndexes[i].PtrIndex(), diffPtrs[i].PtrIndex(),
						sizeof(K) + diffPtrs[i].Length());
				offset += sizeof(K) + diffPtrs[i].Length();

			}
			// Re-sort indexes by key.
			std::sort(m_ptrSecIndexes, m_ptrSecIndexes + m_pairsCount,
					IndexWithLengthLessThan<K, size_t> ());
			m_orderState = OrderState::Combined;
		}
	}
	virtual void Produce()
	{
		ReadFromDisk();
	}
	virtual void Consume()
	{
		m_consumerClass();
	}
	/**
	 * Sort the key-value pairs within this page by key, sort key-value
	 * indexes only rather than move content of key-value pairs.
	 */
	virtual void Sort()
	{
		if (m_orderState == OrderState::None)
		{
			// Sort indexes points to key rather than sort itself.
			std::sort(m_ptrSecIndexes, m_ptrSecIndexes + m_pairsCount,
					IndexWithLengthLessThan<K, size_t> ());
			m_orderState = OrderState::Sorted;
		}
	}

	void RequesPage()
	{
		if (!Page<K, V>::IsWrittenToDisk())
		{
			SetPageBegin(Page<K, V>::m_memPageMaster->AllocatePage());
		}
	}
};


}

#endif /* KEYVALUEPAGEFIXEDKEYVARIANTVALUEIMPL_H_ */
